package cemp.util;

import cemp.common.constant.RedisKeyConstant;
import cemp.common.util.DateUtil;
import cemp.common.util.StockUtil;
import cemp.conf.StockMailSender;
import cemp.entity.StockAll;
import cemp.entity.StockThsGnStocks;
import cemp.redis.util.RedisUtils;
import cemp.service.FetchDataService;
import cemp.service.StockPlateService;
import cemp.service.StockPlateStocksService;
import cemp.service.StockThsGnStocksTotalService;
import com.alibaba.cloud.commons.lang.StringUtils;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AlterConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.ListOffsetsResult;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;


/**
 * @author xshibai
 * @date 2025年07月27日 11:08
 */
@Component
@RequiredArgsConstructor
public class XxlTaskJob {
    private final FetchDataService fetchDataService;
    private final StockPlateService stockPlateService;
    private final StockPlateStocksService stockPlateStocksService;
    private final StockThsGnStocksTotalService stockGnStocksTotalService;
    private final KafkaAdmin kafkaAdmin;
    private final KafkaTemplate<byte[], byte[]> kafkaTemplate;
    private final RedisUtils redisUtils;
    private final StockUtil stockUtil;
    private final StockMailSender stockMailSender;

    @Value("${spring.data.redis.host}")
    private String redisHost;
    @Value("${spring.data.redis.port}")
    private String redisPort;
    @Value("${spring.data.redis.password}")
    private String redisPassword;
    @Value("${stock.mail.from}")
    private String stockFrom;
    @Value("${stock.mail.to}")
    private String stockTo;
//    //用于测试
//    public XxlTaskJob() {
//        fetchDataService = null;
//        stockPlateService = null;
//        stockPlateStocksService = null;
//    }

    private String getTradeDay(){
        String tradeDay = "--";
        tradeDay = (String)redisUtils.get("stock:sta:date");
        return tradeDay;
    }
    private boolean isOpen(){
        if("--".equals(getTradeDay())){
            return false;
        }
        return true;
    }

    private RedisTemplate<String,Object> getTemplateDB3(){
        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration();
        config.setHostName(redisHost);  // 相当于 bootstrap.servers
        config.setPort(Integer.valueOf(redisPort));
        config.setPassword(redisPassword);
        config.setDatabase(3);
        LettuceConnectionFactory factory =  new LettuceConnectionFactory(config);
        factory.afterPropertiesSet();
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);

        // 设置序列化器
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(new StringRedisSerializer());
        template.afterPropertiesSet();
        return template;
    }

    public static void main(String[] args) {
        XxlTaskJob job = new XxlTaskJob(null,null,null,null,null,null,null,null,null);
        RedisTemplate template = job.getTemplateDB3();
        System.out.println("success");
        Set<String> keys =  template.keys("error:stock:kpi:*");
        template.delete(keys);
//        template.opsForValue().set("error:stock:kpi:002648","1");
//        template.opsForValue().set("error:stock:kpi:002649","1");
//        template.opsForValue().increment("error:stock:kpi:002648");
//        template.opsForList().leftPush("list","val3");
//        System.out.println(template.opsForList().size("list"));

    }

    @XxlJob("test")
    public void test() throws Exception {
        System.out.println("test");
        seekToEnd("dfcf_stock_topic",0,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",1,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",2,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",3,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",4,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",5,"python-consumer-group-1");
    }
    /**
     //     * 开盘前 初始化数据 8:00
     //     */
    @XxlJob("beforeOpen")
    public void beforeOpen(){
        //维护上一个交易日
        Object currentRedisDate = redisUtils.get(RedisKeyConstant.STOCK_STA_DATE);
        if(currentRedisDate != null && !"--".equals(currentRedisDate))
            redisUtils.set(RedisKeyConstant.STOCK_STA_YESTERDAY,currentRedisDate.toString());

        //初始化交易日期:查看api 看几天是否开盘 然后维护日期
        String today = cemp.util.DateUtil.getToday();
        Boolean isOpen = stockUtil.isOpen(today);
        RedisTemplate redisDB3 = getTemplateDB3();
        if (isOpen){
            redisUtils.set(RedisKeyConstant.STOCK_STA_DATE,today);
            redisDB3.opsForValue().set(RedisKeyConstant.STOCK_STA_DATE,today);
        }else{
            redisUtils.set(RedisKeyConstant.STOCK_STA_DATE,"--");
            redisDB3.opsForValue().set(RedisKeyConstant.STOCK_STA_DATE,"--");
        }
        redisDB3.delete(RedisKeyConstant.STOCK_COUNT_GN_CURRENT);
        redisDB3.delete(RedisKeyConstant.STOCK_COUNT_GN_TOKEN);
    }

    @XxlJob("endJob")
    public void endJob(){
        if(isOpen()){
            //市盈率写到stockKpiDay
            String tradeDate = getTradeDay();
            List<StockThsGnStocks> list =  fetchDataService.getDistinctList(tradeDate);
            if(list != null &&list.size() > 0){
                list.forEach(gnStock -> fetchDataService.updateKpiDay(gnStock,tradeDate));
            }
            /**删除重复的 gn_stocks**/
            //查询重复成分股中最小的id
//            List<StockThsGnStocks> duplicateList =  fetchDataService.getDuplicate();
            //只保留最小id那一条
//            if(duplicateList != null && duplicateList.size()>0){
//                LambdaQueryWrapper<StockThsGnStocks> lqw = new LambdaQueryWrapper<>();
//                duplicateList.forEach(stock -> fetchDataService.delDuplicate(lqw,stock));
//            }
            /**
             * gn_stocks 写到 gn_stocks_total
             * 逻辑是 只要最近5天出现过就写，并去重
             * 因为gn_stocks 本来就只保留5天数据 所以直接去重即可
             */
            //1. 删除gn_stocks_total
            stockGnStocksTotalService.clear();
            //2. gn_stocks 写到 gn_stocks_total
            stockGnStocksTotalService.mergeData();





        }else{
            System.out.println("今天不开盘");
        }
    }

    /**
     *  通过API 将offset 指向最后一条
     * @param topic
     * @param partition
     * @param groupId
     */
    public void seekToEnd(String topic, int partition, String groupId) {
        Properties props = new Properties();
        props.putAll(kafkaAdmin.getConfigurationProperties());

        try  {
            AdminClient adminClient = AdminClient.create(props);
            // 获取 topic 的 end offsets
            TopicPartition topicPartition = new TopicPartition(topic, partition);
            Map<TopicPartition, OffsetSpec> partitions = Collections.singletonMap(
                    topicPartition, OffsetSpec.latest()
            );

            ListOffsetsResult listOffsetsResult = adminClient.listOffsets(partitions);
            Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> offsets =
                    listOffsetsResult.all().get();

            long endOffset = offsets.get(topicPartition).offset();

            // 修改 consumer group 的 offset
            Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
            offsetMap.put(topicPartition, new OffsetAndMetadata(endOffset));

            AlterConsumerGroupOffsetsResult alterResult = adminClient
                    .alterConsumerGroupOffsets(groupId, offsetMap);

            alterResult.all().get();

            System.out.printf("成功将 topic=%s, partition=%d, group=%s 的 offset 设置为 %d%n",
                    topic, partition, groupId, endOffset);

        } catch (InterruptedException  | ExecutionException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("修改 offset 失败", e);
        }
    }

    /**
     * cron: 0 0/10 16,17,18 * * ?
     * 下午 4点5点6点 每隔10分钟跑一次
     * @throws Exception
     */
    @XxlJob("dfcfStockKpiTecentProduce")
    public void dfcfStockKpiTecentProduce() throws Exception {
        String date = (String)redisUtils.get("stock:sta:date");
        if(StringUtils.isBlank(date) || "--".equals(date)){
            return;
        }
        //查询本批次股票
//        List<StockAll> stocks =  fetchDataService.getDfcfTecentStocks();
        List<StockAll> stocks =  fetchDataService.getDfcfTecentStocks2();
        if (stocks != null && stocks.size() > 0){
            //封装消息对象
            List<JSONObject> msgs = stocks.stream().map(stockAll -> {
                JSONObject msg = new JSONObject();
                msg.put("id",stockAll.getId());
                msg.put("stockCode",stockAll.getStockCode());
                msg.put("stockName",stockAll.getName());
                msg.put("jys",stockAll.getJys().toLowerCase());
                msg.put("staDate",date);
                return msg;
            }).collect(Collectors.toList());
            //生产消息
            msgs.forEach(msg -> {
//            kafkaTemplate.send("dfcf_stock_topic", JSON.toJSONString(msg).getBytes());            //500条数据 每次入列（6个partition）每次只有2个partition收到数据
                kafkaTemplate.send("dfcf_stock_topic_tecent", UUID.randomUUID().toString().getBytes(), JSON.toJSONString(msg).getBytes());     //使用随机KEY 可以让消息入列更均匀
            });

        }

    }

    /**
     * cron: 0 0/10 16,17,18 * * ?
     * 下午 4点5点6点 每隔10分钟跑一次
     * @throws Exception
     */
    @XxlJob("dfcfStockKpiBatchProduce")
    public void dfcfStockKpiBatchProduce() throws Exception {
        //生成当前批次
        RedisTemplate<String,Object> template = getTemplateDB3();
        Long batch = template.opsForValue().increment("batch:stock:kpi");
        //查询所有股票
        List<StockAll> stocks =  fetchDataService.getDfcfBatchStocks(batch.intValue());
        //封装消息对象
        List<JSONObject> msgs = stocks.stream().map(stockAll -> {
            JSONObject msg = new JSONObject();
            msg.put("id",stockAll.getId());
            msg.put("stockCode",stockAll.getStockCode());
            msg.put("stockName",stockAll.getName());
            msg.put("jys",stockAll.getJys().toLowerCase());
            return msg;
        }).collect(Collectors.toList());
        //生产消息
        msgs.forEach(msg -> {
//            kafkaTemplate.send("dfcf_stock_topic", JSON.toJSONString(msg).getBytes());            //500条数据 每次入列（6个partition）每次只有2个partition收到数据
            kafkaTemplate.send("dfcf_stock_topic", UUID.randomUUID().toString().getBytes(), JSON.toJSONString(msg).getBytes());     //使用随机KEY 可以让消息入列更均匀
        });

    }

    /**
     * dfcf_stoock_kpi 一次性投放数据
     * @throws Exception
     */
    @XxlJob("dfcfStockKpiProduce")
    public void dfcfStockKpiProduce() throws Exception {
        //查询所有股票
        List<StockAll> stocks =  fetchDataService.getDfcfStocksAll();
        //封装消息对象
        List<JSONObject> msgs = stocks.stream().map(stockAll -> {
            JSONObject msg = new JSONObject();
            msg.put("id",stockAll.getId());
            msg.put("stockCode",stockAll.getStockCode());
            msg.put("stockName",stockAll.getName());
            msg.put("jys",stockAll.getJys().toLowerCase());
            return msg;
        }).collect(Collectors.toList());
        //生产消息
        msgs.forEach(msg -> {
//            kafkaTemplate.send("dfcf_stock_topic", JSON.toJSONString(msg).getBytes());            //500条数据 每次入列（6个partition）每次只有2个partition收到数据
            kafkaTemplate.send("dfcf_stock_topic", UUID.randomUUID().toString().getBytes(), JSON.toJSONString(msg).getBytes());     //使用随机KEY 可以让消息入列更均匀
        });

    }


    /**
     * 每日任务初始化
     * 对所有job的依赖数据进行初始化
     * mysql:
     * kafka: kafka 对dfcf_stock_topic 在所有partition的offset指向end
     * redis: 对error:stock:xxxx  进行清理
     * @throws Exception
     */
    @XxlJob("daily-job-init")
    public void dailyInit() throws Exception {
        //清理mysql
        //清理kafka
        seekToEnd("dfcf_stock_topic",0,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",1,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",2,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",3,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",4,"python-consumer-group-1");
        seekToEnd("dfcf_stock_topic",5,"python-consumer-group-1");
        //清理redis
        RedisTemplate template = getTemplateDB3();
        Set<String> keys =  template.keys("error:stock:kpi:*");
        template.delete("batch:stock:kpi");     //批次投放计数器
        template.delete(keys);

    }

    /**
     * 收盘后做一些初始化动作
     * @throws Exception
     */
    @XxlJob("fetchData-sta-stocks")
    public void staStocks() throws Exception {
        if(isOpen()){
            RedisTemplate redis = getTemplateDB3();
            redis.delete(redis.keys(RedisKeyConstant.PYTHON_SEL_STOCK_ERROR));
            fetchDataService.maintainMonthly();
        }else{
            System.out.println("今天不开盘");
        }

    }

    /**
     * jenkins任务检验
     * 对gn 和 gn_history两个任务的结果进行检查 如果失败则重复发起任务（最多5次）
     * @throws Exception
     */
    @XxlJob("jenkins-task-review")
    public void jenkinsTaskReview() throws Exception {
        if(isOpen()){
            fetchDataService.jenkinsTaskReview();
        }else{
            System.out.println("今天不开盘");
        }

    }

    @XxlJob("fetchData-msg-inform")
    public void msgInform() throws Exception {
        fetchDataService.cookieExpireNotify();
    }

    @XxlJob("data-sta-gn")
    public void dataStaGn() throws Exception {
        if(isOpen()){
            //比较
        }else{
            System.out.println("今天不开盘");
        }

    }

    @XxlJob("fetchData-sta-kpi")
    public void staKpi() throws Exception {
        if (isOpen()){
            //统计当天的kpi
            String currentDate = DateUtil.date2Str(new Date());
            String stockCode =null;
            fetchDataService.sendBaseHistoryOneByOne(currentDate,currentDate,stockCode);
        }else {
            System.out.println("今天不开盘");
        }
    }

    @XxlJob("fetchData-sta-bk")
    public void staBk() throws Exception {
        stockPlateService.init();
    }


    @XxlJob("fetchData-sta-bk-stocks")
    public void staBkStocks() throws Exception {
        fetchDataService.handlePlateStocks();
    }



}
